package com.mjf.day2

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object MapExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)

    stream.map(_.id).print()

    // 使用自定义类方式重写map方法
    stream.map(new MyMapFunction).print()

    // 使用匿名类的方式重写map方法
    stream.map(new MapFunction[SensorReading, String] {
      override def map(value: SensorReading): String = value.id
    }).print()

    env.execute("MapExample")

  }

  // 输入泛型：SensorReading；输出泛型：String
  class MyMapFunction extends MapFunction[SensorReading, String] {
    override def map(value: SensorReading): String = value.id
  }
}
